package com.xbongbong.paas.elasticsearch.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xbongbong.autoconfigure.data.elasticsearch.ElasticsearchProperties;
import com.xbongbong.i18n.util.I18nMessageUtil;
import com.xbongbong.paas.constant.PaasConstant;
import com.xbongbong.paas.constant.SizeConstant;
import com.xbongbong.paas.elasticsearch.pojo.XbbAggregatedPage;
import com.xbongbong.paas.elasticsearch.pojo.XbbScrolledPage;
import com.xbongbong.paas.enums.BiRelTableEnum;
import com.xbongbong.paas.redis.PaasRedisHelper;
import com.xbongbong.paas.redis.RedisConstant;
import com.xbongbong.paas.redis.RedisPrefixConstant;
import com.xbongbong.paas.toolbox.exception.XbbException;
import com.xbongbong.paas.toolbox.helper.ReflectHelper;
import com.xbongbong.paas.toolbox.util.StringUtil;
import com.xbongbong.pro.config.ProBaseConfig;
import com.xbongbong.pro.constant.I18nStringConstant;
import com.xbongbong.pro.enums.ApplicationNameEnum;
import com.xbongbong.pro.enums.ProVersionEnum;
import com.xbongbong.pro.enums.errorcodes.SystemErrorCodeEnum;
import com.xbongbong.saas.constant.ParameterConstant;
import com.xbongbong.saas.constant.StringConstant;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.common.document.DocumentField;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.query.TermsQueryBuilder;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.domain.PageRequest;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Predicate;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
import static org.elasticsearch.index.query.QueryBuilders.rangeQuery;

/**
 * @author 梁鲁江
 * @version v1.0
 * @date 2019/6/26 19:14
 * @since v1.0
 */
@Configuration
@EnableConfigurationProperties(ElasticsearchProperties.class)
public class XbbElasticsearchRestTemplate {
    private static final Logger LOG = LoggerFactory.getLogger(XbbElasticsearchRestTemplate.class);
    private static String regEx = "(cardinality#)|(min#)|(max#)|(sum#)|(avg#)|(value_count#)|(geo_bounds#)|(geo_centroid#)|(histogram#)|(date_histogram#)|(simple_value#)"
            + "|(auto_date_histogram#)|(sterms#)|(nested#)|(lterms#)|(dterms#)|(missing#)|(reverse_nested#)|(top_hits#)|(children#)";
    private Pattern p = Pattern.compile(regEx);

    /**
     * 指代团队查询中的isMain（是否主负责人）查询
     */
    private static final String IS_MAIN_KEY = "data.num_2";
    /**
     * 客户公海池标记
     */
    private static final String IS_PUBLIC = "data.num_3";
    /**
     * 客户归档标记
     */
    private static final String IS_ARCHIVED = "data.num_4";
    /**
     * convertTerm2Range过滤的key对象，目前：del,flowStatus,logType,data.num_2(isMain),data.num_3,data.num_4
     */
    private static final List<String> TERM_TO_RANGE_KEYS = Arrays.asList(ParameterConstant.DEL, StringConstant.FLOW_STATUS,
            ParameterConstant.LOG_TYPE, IS_MAIN_KEY, IS_PUBLIC, IS_ARCHIVED);
    @Resource
    private ElasticsearchProperties properties;
    /**
     * 通用业务主连接
     */
    private RestClient lowLevelRestClient;

    private RestHighLevelClient restHighLevelClient;
    /**
     * BI业务连接
     */
    private RestClient biLowLevelRestClient;

    private RestHighLevelClient biHighLevelClient;

    private static final String COLON = ":";
    private static final String COMMA = ",";
    private String projectName = ProBaseConfig.getProjectName();
    /**
     * 项目基础配置
     */
    @Resource
    private ProBaseConfig proBaseConfig;

    //pro-web-statistics的时候创建另一个连接池
    @Value("${spring.application.name}")
    private String applicationName;
    @Resource
    private PaasRedisHelper paasRedisHelper;

    @Bean
    public RestClient lowLevelRestClient() throws XbbException {
        RestClient lowLevelRestClient = getLowLevelRestClient(ApplicationNameEnum.PC.getName());
        this.lowLevelRestClient = lowLevelRestClient;
        if (getEnv()) {
            if (ApplicationNameEnum.STATISTICS.getName().equals(applicationName) || ApplicationNameEnum.IMPORT.getName().equals(applicationName)) {
                this.biLowLevelRestClient = getLowLevelRestClient(ApplicationNameEnum.STATISTICS.getName());
            }
        }
        return lowLevelRestClient;
    }

    /**
     * 同一个类的@Bean是按照上下顺序加载的，所以一定要保证lowLevelRestClient在restHighLevelClient钱（也可以使用@DependsOn控制依赖顺序）
     * @return
     * @throws XbbException
     */
    @Bean
    public RestHighLevelClient restHighLevelClient() throws XbbException {
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(getRestClientBuilder(ApplicationNameEnum.PC.getName()));
        this.restHighLevelClient = restHighLevelClient;
        if (getEnv()) {
            if (ApplicationNameEnum.STATISTICS.getName().equals(applicationName) || ApplicationNameEnum.IMPORT.getName().equals(applicationName)) {
                this.biHighLevelClient = new RestHighLevelClient(getRestClientBuilder(ApplicationNameEnum.STATISTICS.getName()));
            }
        }
        return restHighLevelClient;
    }


    /**
     *
     * @return
     */
    private  boolean getEnv(){
        //TODO 上线去除本地的判断
//        boolean env = Objects.equals(projectName, ProVersionEnum.LOCAL.getAlias()) || Objects.equals(projectName, ProVersionEnum.PRD.getAlias());
        boolean env =  Objects.equals(projectName, ProVersionEnum.PRD.getAlias());
        return env;
    }

    /**
     * 当前项目是否是本地运行环境
     * @return true/false
     */
    @SuppressWarnings("AliAccessStaticViaInstance")
    private boolean isLocalEnv() {
        // 必须通过对象访问，保证PostConstruct方法已执行
        return Objects.equals(proBaseConfig.getProjectName(),ProVersionEnum.LOCAL.getAlias());
    }


    public RestClient getRestClient(String indexName) throws XbbException {
        if(!getEnv()){
            return getNormalLowLevelRestClient();
        }
        if(BiRelTableEnum.inHere(indexName)){
            return getBiLevelRestClient();
        }else {
            return getNormalLowLevelRestClient();
        }
    }

    /**
     * 获取通用业务主连接
     * @return
     * @throws XbbException
     */
    private RestClient getNormalLowLevelRestClient() throws XbbException {
        try {
            CloseableHttpAsyncClient closeableHttpAsyncClient = (CloseableHttpAsyncClient) ReflectHelper.valueGet(lowLevelRestClient, "client", false);
            if(lowLevelRestClient==null || closeableHttpAsyncClient==null || ( !closeableHttpAsyncClient.isRunning())){
                this.lowLevelRestClient = getLowLevelRestClient(ApplicationNameEnum.PC.getName());
            }
        } catch (XbbException e) {
           throw e;
        } catch ( Exception e) {
            throw new XbbException(SystemErrorCodeEnum.API_ERROR_100001, I18nMessageUtil.getMessage(I18nStringConstant.STATUS_ERROR), "client");
        }
        return lowLevelRestClient;
    }
    /**
     * 获取BI业务连接
     * @return
     * @throws XbbException
     */
    private RestClient getBiLevelRestClient() throws XbbException {
        try {
            CloseableHttpAsyncClient closeableHttpAsyncClient = (CloseableHttpAsyncClient) ReflectHelper.valueGet(biLowLevelRestClient, "client", false);
            if(biLowLevelRestClient==null || closeableHttpAsyncClient==null || ( !closeableHttpAsyncClient.isRunning())){
                this.biLowLevelRestClient = getLowLevelRestClient(ApplicationNameEnum.STATISTICS.getName());
            }
        } catch (XbbException e) {
           throw e;
        } catch ( Exception e) {
            throw new XbbException(SystemErrorCodeEnum.API_ERROR_100001, I18nMessageUtil.getMessage(I18nStringConstant.STATUS_ERROR), "client");
        }
        return biLowLevelRestClient;
    }

    public RestHighLevelClient getRestHighLevelClient(String indexName) throws XbbException {
        if(!getEnv()){
            return getNormalRestHighLevelClient();
        }
        if(BiRelTableEnum.inHere(indexName)){
            return getBiRestHighLevelClient();
        }else {
            return getNormalRestHighLevelClient();
        }
    }

    private RestHighLevelClient getNormalRestHighLevelClient() throws XbbException {
        RestClient restClient = (RestClient) ReflectHelper.valueGet(restHighLevelClient, "client", true);
        CloseableHttpAsyncClient closeableHttpAsyncClient = (CloseableHttpAsyncClient) ReflectHelper.valueGet(restClient, "client", true);
        if(restHighLevelClient==null || restClient==null || closeableHttpAsyncClient==null || ( !closeableHttpAsyncClient.isRunning())){
            RestClientBuilder restClientBuilder = getRestClientBuilder(ApplicationNameEnum.PC.getName());
            this.restHighLevelClient = new RestHighLevelClient(restClientBuilder);
        }
        return restHighLevelClient;
    }
    private RestHighLevelClient getBiRestHighLevelClient() throws XbbException {
        RestClient restClient = (RestClient) ReflectHelper.valueGet(biHighLevelClient, "client", false);
        CloseableHttpAsyncClient closeableHttpAsyncClient = (CloseableHttpAsyncClient) ReflectHelper.valueGet(restClient, "client", false);
        if(biHighLevelClient==null || restClient==null || closeableHttpAsyncClient==null || ( !closeableHttpAsyncClient.isRunning())){
            this.biHighLevelClient = new RestHighLevelClient(getRestClientBuilder(ApplicationNameEnum.STATISTICS.getName()));
        }
        return biHighLevelClient;
    }

    private RestClient getLowLevelRestClient(String appName) throws XbbException {
        RestClientBuilder restClientBuilder = getRestClientBuilder(appName);
        return restClientBuilder.build();
    }

    private RestClientBuilder getRestClientBuilder(String appName) throws XbbException {
        String clusterNodes = properties.getClusterNodes();
        if(ApplicationNameEnum.STATISTICS.getName().equals(appName) && getEnv()){
            clusterNodes = properties.getBiClusterNodes();
        }
        if (StringUtil.isEmpty(clusterNodes)) {
            LOG.error("请配置cluster-nodes！");
            throw new XbbException(SystemErrorCodeEnum.API_ERROR_100001," Need configure cluster-nodes！");
        }
        String[] nodes = StringUtils.delimitedListToStringArray(clusterNodes, COMMA);
        String finalClusterNodes = clusterNodes;
        List<HttpHost> list = Arrays.stream(nodes).map(node -> {
            String[] segments = StringUtils.delimitedListToStringArray(node, COLON);
            Assert.isTrue(segments.length == 2,
                    () -> String.format("Invalid cluster node %s in %s! Must be in the format host:port!", node, finalClusterNodes));
            String host = segments[0].trim();
            String port = segments[1].trim();
            Assert.hasText(host, () -> String.format("No host name given cluster node %s!", node));
            Assert.hasText(port, () -> String.format("No port given in cluster node %s!", node));
            //            return new HttpHost(toInetAddress(host), Integer.valueOf(port), "http");
            if (!isLocalEnv()) {
                return new HttpHost(toInetAddress(host), 9200, "http");
            }
            // 本地构建直接用配置的host，省去连接nameService，提高效率
            return new HttpHost(host,9200,"http");
        }).collect(Collectors.toList());
        HttpHost[] hostArray = new HttpHost[list.size()];
        RestClientBuilder restClientBuilder = RestClient.builder(list.toArray(hostArray));
        //连接时间   参照 AbstractHttpTransport
        restClientBuilder.setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder.
                //10000
                setConnectTimeout(60000)
                //10000
                .setConnectionRequestTimeout(60000)
                //600000
                .setSocketTimeout(60000))
                // 默认值  DEFAULT_MAX_RETRY_TIMEOUT_MILLIS  30000
//                .setMaxRetryTimeoutMillis(60000)
        ;
        //Xpack
        if (properties.getXpack() != null && StringUtil.isNotEmpty(properties.getXpack().getUser()) && StringUtil.isNotEmpty(properties.getXpack().getPassword())) {
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY,
                    new UsernamePasswordCredentials(properties.getXpack().getUser(), properties.getXpack().getPassword()));
            restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
                    //是整个连接池的大小   AbstractHttpTransport  1000 （默认30，暂时设置成数据库连接池一致50）
                    .setMaxConnTotal(10000)
                    //是单个路由连接的最大数  AbstractHttpTransport  500
                    //默认  RestClientBuilder.DEFAULT_MAX_CONN_PER_ROUTE  默认10
                    .setMaxConnPerRoute(50)
                    .setDefaultCredentialsProvider(credentialsProvider)
                    .setDefaultIOReactorConfig(
                            IOReactorConfig.custom()
                                    .build()));
        }
        return restClientBuilder;
    }

    /**
     * Description: 只查列表没有总数
     * @param pageRequest
     * @param searchRequest
     * @param clazz
     * @return com.xbongbong.paas.elasticsearch.pojo.XbbAggregatedPage<T>
     * @throws
     * @author 魏荣杰
     * @date 2020/11/25 14:21
     * @since
     */
    public  <T> XbbAggregatedPage<T> queryForPagesWithoutCount(PageRequest pageRequest, SearchRequest searchRequest, Class<T> clazz) throws XbbException {
        XbbAggregatedPage<T> xbbAggregatedPage = new XbbAggregatedPage<>();
        try {
            SearchSourceBuilder searchSourceBuilder = searchRequest.source();
            SearchResponse searchResponse = search(searchRequest, searchSourceBuilder,false);
            SearchHits hits = searchResponse.getHits();
            SearchHit[] arrayHits = hits.getHits();
            List<T> resultList = new ArrayList<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
            for (SearchHit hit : arrayHits) {
                JSONObject jsonObject = JSON.parseObject(hit.getSourceAsString());
                resultList.add(jsonObject.toJavaObject(clazz));
            }
            xbbAggregatedPage.setContent(resultList);
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
        return xbbAggregatedPage;
    }

    /**
     * 既要列表又要总数
     * @param pageRequest
     * @param searchRequest
     * @param clazz
     * @param <T>
     * @return
     * @throws XbbException
     */
    public  <T> XbbAggregatedPage<T> queryForPages(PageRequest pageRequest, SearchRequest searchRequest, Class<T> clazz) throws XbbException {
        XbbAggregatedPage<T> xbbAggregatedPage = new XbbAggregatedPage<>();
        try {
            SearchSourceBuilder searchSourceBuilder = searchRequest.source();
            SearchResponse searchResponse = search(searchRequest, searchSourceBuilder);
            SearchHits hits = searchResponse.getHits();
            long count = hits.getTotalHits().value;
            SearchHit[] arrayHits = hits.getHits();
            List<T> resultList = new ArrayList<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
            for (SearchHit hit : arrayHits) {
                JSONObject jsonObject = JSON.parseObject(hit.getSourceAsString());
                //如果hit.getFields.values不为空说明有scriptFields
                Map<String, DocumentField> fields = hit.getFields();
                if (!fields.isEmpty()) {
                    for (String s : fields.keySet()) {
                        jsonObject.put(s,fields.get(s).getValues().get(0));
                    }
                }
                resultList.add(JSON.parseObject(hit.getSourceAsString(), clazz));
            }
            if (arrayHits.length>0) {
                xbbAggregatedPage.setSortValue(arrayHits[arrayHits.length-1].getSortValues());
            }
            xbbAggregatedPage.setContent(resultList);
            xbbAggregatedPage.setTotalElements(count);
            xbbAggregatedPage.setAggregations(searchResponse.getAggregations());
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
        return xbbAggregatedPage;
    }

    public  <T> XbbAggregatedPage<T> queryForPages( SearchRequest searchRequest, Class<T> clazz) throws XbbException {
        XbbAggregatedPage<T> xbbAggregatedPage = new XbbAggregatedPage<>();
        try {
            SearchSourceBuilder searchSourceBuilder = searchRequest.source();
            SearchResponse searchResponse = search(searchRequest, searchSourceBuilder);
            SearchHits hits = searchResponse.getHits();
            SearchHit[] arrayHits = hits.getHits();
            List<T> resultList = new ArrayList<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
            for (SearchHit hit : arrayHits) {
                JSONObject jsonObject = JSON.parseObject(hit.getSourceAsString());
                resultList.add(jsonObject.toJavaObject(clazz));
            }
            xbbAggregatedPage.setContent(resultList);
            long count = hits.getTotalHits().value;
            xbbAggregatedPage.setTotalElements(count);
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
        return xbbAggregatedPage;
    }


    public  <T> XbbAggregatedPage<T> queryOfDistributor( SearchRequest searchRequest, Class<T> clazz) throws XbbException {
        XbbAggregatedPage<T> xbbAggregatedPage = new XbbAggregatedPage<>();
        try {
            SearchSourceBuilder searchSourceBuilder = searchRequest.source();
            SearchResponse searchResponse = search(searchRequest, searchSourceBuilder);
            SearchHits hits = searchResponse.getHits();
            SearchHit[] arrayHits = hits.getHits();
            List<T> resultList = new ArrayList<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
            for (SearchHit hit : arrayHits) {
                JSONObject jsonObject = JSON.parseObject(hit.getSourceAsString());
                resultList.add(jsonObject.toJavaObject(clazz));
            }
            xbbAggregatedPage.setContent(resultList);
            CountRequest countRequest = new CountRequest(searchRequest.indices()[0]);
            countRequest.query(searchRequest.source().query());
            long count = hits.getTotalHits().value;
            xbbAggregatedPage.setTotalElements(count);
            Aggregations aggregations = searchResponse.getAggregations();
            xbbAggregatedPage.setAggregations(aggregations);
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
        return xbbAggregatedPage;
    }

    public  <T> SearchResponse searchResponse(PageRequest pageRequest, SearchRequest searchRequest, Class<T> clazz) throws XbbException {
        try {
            printEsInfo(searchRequest);
            SearchSourceBuilder searchSourceBuilder = searchRequest.source();
            return search(searchRequest, searchSourceBuilder);
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
    }

    public SearchResponse search(SearchRequest searchRequest) throws XbbException {
        try {
            return search(searchRequest,searchRequest.source());
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
    }

    /**
     * 获取聚合结果的responseJson
     * @param searchRequest
     * @return
     * @throws XbbException
     */
    public JSONObject getAggResponseJson(SearchRequest searchRequest) throws XbbException {
        try {
            SearchResponse searchResponse = search(searchRequest);
            JSONObject jsonObject = JSONObject.parseObject(searchResponse.toString());
            JSONObject totalJson = new JSONObject();
            totalJson.put("total",searchResponse.getHits().getTotalHits().value);
            JSONObject returnJson = new JSONObject();
            returnJson.put("hits",totalJson);
            JSONObject aggJson = jsonObject.getJSONObject("aggregations");
            if (aggJson != null) {
                Matcher m = p.matcher(aggJson.toString());
                String result = m.replaceAll("");
                returnJson.put("aggregations",JSONObject.parseObject(result));
            }
            return returnJson;
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
    }

    /**
     * @Description 经销商图表中心用到，封装订单关联商品数据
     * @param searchRequest
     * @return JSONObject
     * @author weiming
     * @Date  2021/1/21 20:17
     * @since
     **/
    public JSONObject getAggResponse4Product(SearchRequest searchRequest) throws XbbException {
        try {
            SearchResponse searchResponse = search(searchRequest);
            JSONObject jsonObject = JSONObject.parseObject(searchResponse.toString());
            JSONObject totalJson = new JSONObject();
            SearchHits hits = searchResponse.getHits();
            SearchHit[] arrayHits = hits.getHits();
            //封装Map<skuid,姓名+规格信息>
            Map<Integer, JSONObject> topSkuData = new HashMap<>();
            for (SearchHit hit : arrayHits) {
                JSONObject tempObject = JSON.parseObject(hit.getSourceAsString());
                //把tempObject的data取出
                JSONObject dataObject = tempObject.getJSONObject("data");
                //skuid
                Integer long_2 = dataObject.getInteger("long_2");
                //规格信息
                String text_26 = dataObject.getString("text_26");
                //sku名称
                String text_23 = dataObject.getString("text_23");
                Long num_32 = dataObject.getLong("num_32");
                JSONObject object = new JSONObject();
                object.put("productName",text_23);
                object.put("specText",text_26);
                object.put("productNum",num_32);
                //把数据存到map中
                topSkuData.put(long_2,object);
                //resultList.add(jsonObject.toJavaObject(clazz));
            }
            totalJson.put("total",searchResponse.getHits().getTotalHits().value);
            JSONObject returnJson = new JSONObject();
            returnJson.put("hits",totalJson);
            returnJson.put("data",topSkuData);
            JSONObject aggJson = jsonObject.getJSONObject("aggregations");
            if (aggJson != null) {
                Matcher m = p.matcher(aggJson.toString());
                String result = m.replaceAll("");
                returnJson.put("aggregations",JSONObject.parseObject(result));
            }
            return returnJson;
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
    }


    public <T> List<T> queryForList(PageRequest pageRequest, SearchRequest searchRequest, Class<T> clazz) throws XbbException {
        List<T> resultList = new ArrayList<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
        try {
            SearchSourceBuilder searchSourceBuilder = searchRequest.source();
            SearchResponse searchResponse = search(searchRequest, searchSourceBuilder,false);
            SearchHit[] hits = searchResponse.getHits().getHits();
            for (SearchHit hit : hits) {
                JSONObject jsonObject = JSON.parseObject(hit.getSourceAsString());
                resultList.add(jsonObject.toJavaObject(clazz));
            }
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
        return resultList;
    }

    public <T> List<T> queryForList( SearchRequest searchRequest, Class<T> clazz) throws XbbException {
        List<T> resultList = new ArrayList<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
        try {
            SearchSourceBuilder searchSourceBuilder = searchRequest.source();
            SearchResponse searchResponse = search(searchRequest, searchSourceBuilder,false);
            SearchHit[] hits = searchResponse.getHits().getHits();
            for (SearchHit hit : hits) {
                JSONObject jsonObject = JSON.parseObject(hit.getSourceAsString());
                resultList.add(jsonObject.toJavaObject(clazz));
            }
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
        return resultList;
    }

    @Deprecated
    public <T>long count( SearchRequest searchRequest,  Class<T> clazz) throws XbbException {
        CountRequest countRequest = new CountRequest(searchRequest.indices()[0]);
        countRequest.query(searchRequest.source().query());
        return count(countRequest);
    }

    public long count(CountRequest countRequest) throws XbbException {
        try {
            SearchSourceBuilder searchSourceBuilder = countRequest.source();
            //将Del Term查询转化为Range查询
            convertTerm2Range(searchSourceBuilder, TERM_TO_RANGE_KEYS);
            printEsInfo(countRequest.indices()[0],countRequest.query());
            CountResponse countResponse = getRestHighLevelClient(countRequest.indices()[0]).count(countRequest, RequestOptions.DEFAULT);
            return countResponse.getCount();
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
    }

    public  <T> XbbScrolledPage<T> startScroll(long scrollTimeInMillis, SearchRequest searchRequest, Class<T> clazz) throws XbbException {
        XbbScrolledPage<T> xbbScrolledPage  = new XbbScrolledPage<>();
        try {
            printEsInfo(searchRequest);
            // If set, will enable scrolling of the search request for the specified timeout.
            searchRequest.scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
            SearchSourceBuilder sourceBuilder = searchRequest.source();
            int from = sourceBuilder.from();
            int size = sourceBuilder.size();
            from = from == 0 ? 1 : from;
            size = size == 0 ? SizeConstant.PAGE_SIZE : size;
            int page = from / size;
            page = page < 0 ? 0 : page;
            EsUtil.setPage(sourceBuilder, page,size);
            SearchResponse searchResponse = search(searchRequest, sourceBuilder);
            xbbScrolledPage.setScrollId(searchResponse.getScrollId());
            SearchHits hits = searchResponse.getHits();
            SearchHit[] arrayHits = hits.getHits();
            List<T> resultList = new ArrayList<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
            for (SearchHit hit : arrayHits) {
                JSONObject jsonObject = JSON.parseObject(hit.getSourceAsString());
                Map<String, DocumentField> computedFields = hit.getFields();
                if (null != computedFields){
                    computedFields.forEach((k,v) ->{
                        jsonObject.put(k,v.getValue());
                    });
                }
                resultList.add(jsonObject.toJavaObject(clazz));
            }
            if(resultList.size() > 0){
                xbbScrolledPage.setHasContent(true);
            }
            xbbScrolledPage.setContent(resultList);
            xbbScrolledPage.setTotalElements(hits.getTotalHits().value);
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
        return xbbScrolledPage;
    }
    public  <T> XbbScrolledPage<T> continueScroll(long scrollTimeInMillis, XbbScrolledPage scrolledPage, Class<T> clazz,String indexName) throws XbbException {
        XbbScrolledPage<T> xbbScrolledPage = new XbbScrolledPage<>();
        try {
            String scrollId = scrolledPage.getScrollId();
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            final Scroll scroll = new Scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
            scrollRequest.scroll(scroll);
            SearchResponse searchResponse = getRestHighLevelClient(indexName).scroll(scrollRequest, RequestOptions.DEFAULT);
            SearchHits hits = searchResponse.getHits();
            SearchHit[] arrayHits = hits.getHits();
            List<T> resultList = new ArrayList<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
            for (SearchHit hit : arrayHits) {
                JSONObject jsonObject = JSON.parseObject(hit.getSourceAsString());
                resultList.add(jsonObject.toJavaObject(clazz));
            }
            if(resultList.size() > 0){
                xbbScrolledPage.setHasContent(true);
            }
            xbbScrolledPage.setScrollId(searchResponse.getScrollId());
            xbbScrolledPage.setContent(resultList);
            xbbScrolledPage.setTotalElements(hits.getTotalHits().value);
            return xbbScrolledPage;
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
    }

    public void clearScroll(String scrollId,String indexName)  throws XbbException {
        try {
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            //TODO  如果有需求可以解析返回值
//            ClearScrollResponse clearScrollResponse =
            getRestHighLevelClient(indexName).clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
    }

    /**
     * 不建议使用
     * 获取符合条件的所有数据，如有分页获取需求请调用queryForList，禁止在循环内使用
     * @param clazz
     * @param searchRequest
     * @param scrollTimeInMillis
     * @param <T>
     * @return
     * @throws XbbException
     */
    public <T>List<T> findScrollList(Class<T> clazz, SearchRequest searchRequest, long scrollTimeInMillis) throws XbbException {
        List<T> resultList = new ArrayList<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
        try {
            // If set, will enable scrolling of the search request for the specified timeout.
            searchRequest.scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
            SearchSourceBuilder sourceBuilder = searchRequest.source();
            int from = sourceBuilder.from();
            int size = sourceBuilder.size();
            from = from == 0 ? 1 : from;
            size = size == 0 ? SizeConstant.PAGE_SIZE : size;
            int page = from / size;
            page = Math.max(page, 0);
            EsUtil.setPage(sourceBuilder, page,size);
            SearchResponse searchResponse = search(searchRequest, sourceBuilder,false);
            String scrollId = searchResponse.getScrollId();
            SearchHits hits = searchResponse.getHits();
            SearchHit[] arrayHits = hits.getHits();
            final Scroll scroll = new Scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
            String index = searchRequest.indices()[0];
            while (arrayHits != null && arrayHits.length > 0) {
                for (SearchHit hit : arrayHits) {
                    JSONObject jsonObject = JSON.parseObject(hit.getSourceAsString());
                    resultList.add(jsonObject.toJavaObject(clazz));
                }
                if(arrayHits.length < size){
                    break;
                }
                SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                scrollRequest.scroll(scroll);
                searchResponse = getRestHighLevelClient(index).scroll(scrollRequest, RequestOptions.DEFAULT);
                scrollId = searchResponse.getScrollId();
                arrayHits = searchResponse.getHits().getHits();
            }
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            //TODO  如果有需求可以解析返回值
//            ClearScrollResponse clearScrollResponse =
            getRestHighLevelClient(index).clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
        return resultList;
    }

    /**
     * 导出的5W条限制，滚动读取
     * @param clazz
     * @param searchRequest
     * @return
     * @throws XbbException
     */
    public List getScrollListForExport(Class clazz, SearchRequest searchRequest) throws XbbException {
        long scrollTimeInMillis = SizeConstant.SCROLL_TIME_IN_MILLIS;
        XbbScrolledPage scrolledPage = startScroll(scrollTimeInMillis, searchRequest, clazz);
        List list = new ArrayList();
        for (int i = 0; i < 5; i++) {
            if (scrolledPage.isHasContent()) {
                List content = scrolledPage.getContent();
                list.addAll(content);
                scrolledPage = continueScroll(scrollTimeInMillis, scrolledPage, clazz,searchRequest.indices()[0]);
            } else {
                break;
            }
        }
        clearScroll(scrolledPage.getScrollId(),searchRequest.indices()[0]);
        return list;
    }


    public Aggregations  query(SearchRequest searchRequest) throws XbbException {
        try {
            SearchSourceBuilder searchSourceBuilder = searchRequest.source();
            //将Del Term查询转化为Range查询
            convertTerm2Range(searchSourceBuilder, TERM_TO_RANGE_KEYS);
            printEsInfo(searchRequest);
            SearchResponse searchResponse = getRestHighLevelClient(searchRequest.indices()[0]).search(searchRequest, RequestOptions.DEFAULT);
            return searchResponse.getAggregations();
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
    }

    /**
     * 删除索引（慎用）
     * @param indexName
     * @return
     */
    public boolean deleteIndex(String indexName) throws XbbException {
        DeleteIndexRequest request = new DeleteIndexRequest(indexName);
        try {
            AcknowledgedResponse deleteIndexResponse = getRestHighLevelClient(indexName).indices().delete(request, RequestOptions.DEFAULT);
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100073.getMsg(),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100073);
        }
        return true;
    }

    private static InetAddress toInetAddress(String host) {
        try {
            return InetAddress.getByName(host);
        } catch (UnknownHostException e) {
            throw new IllegalArgumentException(e);
        }
    }

    private SearchResponse search(SearchRequest searchRequest, SearchSourceBuilder searchSourceBuilder) throws XbbException,Exception {
        return search(searchRequest, searchSourceBuilder, true);
    }

    private SearchResponse search(SearchRequest searchRequest, SearchSourceBuilder searchSourceBuilder, boolean trackTotalHits) throws XbbException {
        //将Del Term查询转化为Range查询
        convertTerm2Range(searchSourceBuilder, TERM_TO_RANGE_KEYS);
        if(trackTotalHits){
            searchSourceBuilder.trackTotalHits(trackTotalHits);
        }
        SearchResponse searchResponse;
        try {
            printEsInfo(searchRequest);
            searchResponse = getRestHighLevelClient(searchRequest.indices()[0]).search(searchRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            LOG.error(SystemErrorCodeEnum.API_ERROR_100071.getMsg() +"=========="+ getEsLog(searchRequest),e);
            throw  new XbbException(SystemErrorCodeEnum.API_ERROR_100071);
        }
        return searchResponse;
    }

    public void setPrintLog(boolean printLog){
        paasRedisHelper.setValue(RedisPrefixConstant.PRINT_ES_LOG_KEY,applicationName,printLog, RedisConstant.LONG_DURATION);
    }
    /**
     * 查询日志默认关闭
     */
    public boolean getPrintLog (){
        String booleanStr;
        try {
            booleanStr = paasRedisHelper.getValue(RedisPrefixConstant.PRINT_ES_LOG_KEY, applicationName);
        } catch (Exception e) {
            LOG.error("",e);
            return  false;
        }
        if(StringUtil.isEmpty(booleanStr)){
            return  false;
        }
        try {
            return Boolean.valueOf(booleanStr);
        } catch (Exception e) {
            LOG.error("",e);
            return false;
        }
    }


    private void printEsInfo(SearchRequest searchRequest) {
        if(!getPrintLog()){
            return;
        }
        StringBuilder stringBuilder = getEsLog(searchRequest); 
        LOG.info(stringBuilder.toString());
    }

    private StringBuilder getEsLog(SearchRequest searchRequest) {
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("\n GET  " );
        stringBuilder.append(searchRequest.indices()[0]);
        stringBuilder.append("/_search\n" );
        try {
            stringBuilder.append(searchRequest.source().toString());
        } catch (Exception e) {
           LOG.error("",e);
        }
        return stringBuilder;
    }

    private void printEsInfo(String index,QueryBuilder query) {
        if(!getPrintLog()){
            return;
        }
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("\n GET  " );
        stringBuilder.append(index);
        stringBuilder.append("/_count\n" );
        stringBuilder.append("{\"query\":");
        stringBuilder.append(query);
        stringBuilder.append("}");
        LOG.info(stringBuilder.toString());
    }

    /**
     * 目前只用于 del 和 flowStatus,其他字段需要转化调用需告知 @何卡特
     * 增加data.num_2字段，用以处理团队查询中is_main查询逻辑
     * terms只替换项数少于4项的场景，否则可能导致语句膨胀
     * 解析sourceBuilder第一层级，找出其中的数值型key对应的Term和Terms语句，并将其替换为Range语句；
     * 该动作应用于es mapping中没有设置keyword的数值型字段，且该字段需要执行term逻辑，
     * 若是新索引建议给需要频繁做term或terms的数值字段建keyword。
     * 本解析方法用于查询优化，可能只是暂时的。
     * 解释：rangeQuery的结果集非常巨大，则构造bitset的过程会大大延缓scorer对象的构造过程，造成结果合并过程缓慢。
     * 从ES5.4开始，引入了indexOrDocValuesQuery作为对RangeQuery的优化。
     * indexOrDocValuesQuery将Range查询过程中的顺序访问任务扔给Block k-d Tree索引，将随机访任务交给doc values。
     * 值得注意的是目前这个优化只针对RangeQuery。对于TermQuery，因为实际的复杂性，还未做类似的优化，
     * 也就导致对于数值型字段，Term和Range Query的性能差异极大。
     * @param sourceBuilder SearchSourceBuilder
     * @param keys 需要转化的keys,目前只用于 del 和 flowStatus,其他字段需要转化调用需告知 @何卡特
     * @author kaka
     * @date 2022-3-22
     */
    private static void convertTerm2Range(SearchSourceBuilder sourceBuilder, List<String> keys) {
        QueryBuilder queryBuilder = sourceBuilder.query();
        if (!(queryBuilder instanceof BoolQueryBuilder)) {
            return;
        }
        // 替换规则，key包含在keys中,且是term查询，或者是terms查询并且项数小于等于4项；
        Predicate<? super QueryBuilder> qbPredicate = qb -> (qb instanceof TermQueryBuilder && keys.contains(((TermQueryBuilder)qb).fieldName()))
                || (qb instanceof TermsQueryBuilder && keys.contains(((TermsQueryBuilder)qb).fieldName()) && ((TermsQueryBuilder)qb).values().size() <= 4);
        List<QueryBuilder> queryBuilders = ((BoolQueryBuilder)queryBuilder).filter();
        List<QueryBuilder> replaceQuery = queryBuilders.stream()
                .filter(qbPredicate)
                .map(qb -> {
                    if (qb instanceof TermQueryBuilder) {
                        String key = ((TermQueryBuilder) qb).fieldName();
                        TermQueryBuilder tqb = (TermQueryBuilder)qb;
                        Object value = tqb.value();
                        return rangeQuery(key).gte(value).lte(value);
                    } else {
                        // filter中过滤后只有TermQueryBuilder和TermsQueryBuilder，所以else只有TermsQueryBuilder
                        String key = ((TermsQueryBuilder) qb).fieldName();
                        // 保险起见，data.num_2先只替换term，暂不替换terms
                        if (Objects.equals(IS_MAIN_KEY, key)) {
                            return qb;
                        }
                        TermsQueryBuilder termsQB = (TermsQueryBuilder)qb;
                        List<Object> values = termsQB.values();
                        // 优化Terms查询只有一个项的情况，转化为Term查询
                        if (values.size() == 1) {
                            Object value = values.get(0);
                            return rangeQuery(key).gte(value).lte(value);
                        } else {
                            // Terms查询且项数大于等于2的情况
                            BoolQueryBuilder termsBQB = boolQuery();
                            List<QueryBuilder> termsShouldQuery = termsBQB.should();
                            for (Object value : values) {
                                termsShouldQuery.add(rangeQuery(key).gte(value).lte(value));
                            }
                            return termsBQB;
                        }
                    }
                }).collect(Collectors.toList());
        queryBuilders.removeIf(qbPredicate);
        queryBuilders.addAll(replaceQuery);
    }
}
